[アップデート] Amazon OpenSearch Ingestion が Amazon Kinesis Data Streams からのデータの取り込みをサポートしました

export const handler = async (event) => {
  console.error("エラーです");
  return "エラーです";
};

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "logs.ap-northeast-1.amazonaws.com"
      },
      "Action": "sts:AssumeRole",
      "Condition": {
        "StringLike": {
          "aws:SourceArn": "arn:aws:logs:ap-northeast-1:<account-id>:*"
        }
      }
    }
  ]
}

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": "kinesis:PutRecord",
            "Resource": "arn:aws:kinesis:ap-northeast-1:<account-id>:stream/kinesis-data-streams-to-open-search"
        }
    ]
}

{
  "name": "opensearch-vpc-cdk",
  "version": "0.1.0",
  "bin": {
    "opensearch-vpc-cdk": "bin/opensearch-vpc-cdk.js"
  },
  "scripts": {
    "build": "tsc",
    "watch": "tsc -w",
    "test": "jest",
    "cdk": "cdk"
  },
  "devDependencies": {
    "@types/jest": "^29.5.14",
    "@types/node": "22.9.0",
    "aws-cdk": "2.167.1",
    "jest": "^29.7.0",
    "ts-jest": "^29.2.5",
    "ts-node": "^10.9.2",
    "typescript": "~5.6.3"
  },
  "dependencies": {
    "@aws-cdk/aws-lambda-python-alpha": "^2.167.1-alpha.0",
    "@aws-sdk/client-iam": "^3.693.0",
    "aws-cdk-lib": "2.167.1",
    "constructs": "^10.4.2",
    "source-map-support": "^0.5.21"
  }
}

#!/usr/bin/env node
import "source-map-support/register";
import * as cdk from "aws-cdk-lib";
import { OpensearchVpcCdkStack } from "../lib/opensearch-vpc-cdk-stack";

const app = new cdk.App();
new OpensearchVpcCdkStack(app, "OpensearchVpcCdkStack", {});

import * as cdk from "aws-cdk-lib";
import { CfnOutput, RemovalPolicy } from "aws-cdk-lib";
import {
  BastionHostLinux,
  BlockDeviceVolume,
  MachineImage,
  Peer,
  Port,
  SecurityGroup,
  Vpc,
} from "aws-cdk-lib/aws-ec2";
import {
  AnyPrincipal,
  CfnServiceLinkedRole,
  PolicyStatement,
} from "aws-cdk-lib/aws-iam";
import { Domain, EngineVersion } from "aws-cdk-lib/aws-opensearchservice";
import { Construct } from "constructs";
import { IAMClient, ListRolesCommand } from "@aws-sdk/client-iam";

const iam = new IAMClient({});

export class OpensearchVpcCdkStack extends cdk.Stack {
  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
    super(scope, id, props);

    // VPC
    const vpc = new Vpc(this, "Vpc", {});

    // Security Group
    const bastionSecurityGroup = new SecurityGroup(
      this,
      "BastionSecurityGroup",
      {
        vpc: vpc,
        allowAllOutbound: true,
        securityGroupName: "BastionSecurityGroup",
      }
    );

    const opensearchSecurityGroup = new SecurityGroup(
      this,
      "OpensearchSecurityGroup",
      {
        vpc: vpc,
        securityGroupName: "OpensearchSecurityGroup",
      }
    );

    opensearchSecurityGroup.addIngressRule(bastionSecurityGroup, Port.tcp(443));

    // Service-linked role that Amazon OpenSearch Service will use
    (async () => {
      const response = await iam.send(
        new ListRolesCommand({
          PathPrefix: "/aws-service-role/opensearchservice.amazonaws.com/",
        })
      );

      // Only if the role for OpenSearch Service doesn't exist, it will be created.
      if (response.Roles && response.Roles?.length == 0) {
        new CfnServiceLinkedRole(this, "OpensearchServiceLinkedRole", {
          awsServiceName: "es.amazonaws.com",
        });
      }
    })();

    // Bastion host to access Opensearch Dashboards
    new BastionHostLinux(this, "BastionHost", {
      vpc,
      securityGroup: bastionSecurityGroup,
      machineImage: MachineImage.latestAmazonLinux2023(),
      blockDevices: [
        {
          deviceName: "/dev/xvda",
          volume: BlockDeviceVolume.ebs(10, {
            encrypted: true,
          }),
        },
      ],
    });

    // OpenSearch domain
    const domain = new Domain(this, "Domain", {
      version: EngineVersion.OPENSEARCH_2_15,
      nodeToNodeEncryption: true,
      enforceHttps: true,
      encryptionAtRest: {
        enabled: true,
      },
      vpc: vpc,
      capacity: {
        dataNodes: 2,
      },
      removalPolicy: RemovalPolicy.DESTROY,
      zoneAwareness: {
        enabled: true,
      },
      securityGroups: [opensearchSecurityGroup],
    });

    domain.addAccessPolicies(
      new PolicyStatement({
        principals: [new AnyPrincipal()],
        actions: ["es:ESHttp*"],
        resources: [domain.domainArn + "/*"],
      })
    );
  }
}

npm ci
npx cdk bootstrap
npx cdk deploy

aws ssm start-session --target <EC2のインスタンスID> --document-name AWS-StartPortForwardingSessionToRemoteHost --parameters '{"portNumber":["443"],"localPortNumber":["8157"], "host":["*******.ap-northeast-1.es.amazonaws.com"]}'


Starting session with SessionId: **************
Port 8157 opened for sessionId **************
Waiting for connections...

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Principal": {
        "Service": "osis-pipelines.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }
  ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "allowReadFromStream",
      "Effect": "Allow",
      "Action": [
        "kinesis:DescribeStream",
        "kinesis:DescribeStreamConsumer",
        "kinesis:DescribeStreamSummary",
        "kinesis:GetRecords",
        "kinesis:GetShardIterator",
        "kinesis:ListShards",
        "kinesis:ListStreams",
        "kinesis:ListStreamConsumers",
        "kinesis:RegisterStreamConsumer",
        "kinesis:SubscribeToShard"
      ],
      "Resource": ["arn:aws:kinesis:ap-northeast-1:<account-id>:stream/*"]
    },
    {
      "Sid": "allowAccessToOS",
      "Effect": "Allow",
      "Action": ["es:DescribeDomain", "es:ESHttp*"],
      "Resource": [
        "arn:aws:es:ap-northeast-1:<account-id>:domain/domain66ac69e0-xwfaqbkvba0m",
        "arn:aws:es:ap-northeast-1:<account-id>:domain/domain66ac69e0-xwfaqbkvba0m/*"
      ]
    }
  ]
}

version: "2"
kinesis-pipeline:
  source:
    kinesis-data-streams:
      acknowledgments: true
      codec:
        # Kinesisレコードが集約されているかどうかに基づいて、json、newline、またはndjsonコーデックを選択できます。
        # JSONコーデックは、ネストされたCloudWatchイベントを個々のログエントリに解析し、OpenSearchにドキュメントとして書き込みます。
        json:
          key_name: "logEvents"
          # これらのキーには、CloudWatchサブスクリプションフィルターによって送信されたメタデータが含まれています。
          # 個々のログイベントに加えて:
          # https://docs.aws.amazon.com/AmazonCloudWatch/latest/logs/SubscriptionFilters.html#DestinationKinesisExample
          # include_keys: [ 'owner', 'logGroup', 'logStream' ]
        newline:
      streams:
        - stream_name: "kinesis-data-streams-to-open-search"
          # ストリームの開始から取り込みを開始する場合はこれを有効にします。
          initial_position: "EARLIEST"
          # checkpoint_interval: "PT5M"
          # CloudWatchの場合、圧縮は常にgzipになりますが、他のソースの場合は異なります:
          compression: "gzip"
      aws:
        # KDSへのアクセスを持つRole ARNを提供します。このロールはosis-pipelines.amazonaws.comとの信頼関係を持つ必要があります。
        sts_role_arn: "arn:aws:iam::<account-id>:role/kds-to-es-integration-pipeline-role"
        # データストリームのリージョンを提供します。
        region: "ap-northeast-1"

  processor:
    - rename_keys:
        entries:
          # CloudWatchのタイムスタンプを観測タイムスタンプとして含めます - ログが生成され、CloudWatchに送信された時間:
          - from_key: "timestamp"
            to_key: "observed_timestamp"
    - date:
        # OSIがログイベントを処理した現在のタイムスタンプを含めます:
        from_time_received: true
        destination: "processed_timestamp"
    - add_entries:
        entries:
          # SS4O共通ログフィールドをサポートします (https://opensearch.org/docs/latest/observing-your-data/ss4o/)
          - key: "cloud/provider"
            value: "aws"
          - key: "cloud/account/id"
            format: "${owner}"
          - key: "cloud/region"
            value: "ap-northeast-1"
          - key: "aws/cloudwatch/log_group"
            format: "${logGroup}"
          - key: "aws/cloudwatch/log_stream"
            format: "${logStream}"
          # data_streamのデフォルト値を含めます:
          - key: "data_stream/namespace"
            value: "default"
          - key: "data_stream/type"
            value: "logs"
          - key: "data_stream/dataset"
            value: "general"
          # このログイベントを含むソースKinesisメッセージに関するメタデータを含めます:
          - key: "aws/kinesis/stream_name"
            value_expression: 'getMetadata("stream_name")'
          - key: "aws/kinesis/partition_key"
            value_expression: 'getMetadata("partition_key")'
          - key: "aws/kinesis/sequence_number"
            value_expression: 'getMetadata("sequence_number")'
          - key: "aws/kinesis/sub_sequence_number"
            value_expression: 'getMetadata("sub_sequence_number")'
    - add_entries:
        entries:
          # ログイベントのコンテキストに基づいてdata_streamフィールドを更新します - この場合、ソース（CloudTrailまたはLambda）によってログイベントを分類します。
          # ビジネスまたはアプリケーションコンテキストによってログを分類するための追加のロジックを追加できます:
          - key: "data_stream/dataset"
            value: "cloudtrail"
            add_when: 'contains(/logGroup, "cloudtrail") or contains(/logGroup, "CloudTrail")'
            overwrite_if_key_exists: true
          - key: "data_stream/dataset"
            value: "lambda"
            add_when: 'contains(/logGroup, "/aws/lambda/")'
            overwrite_if_key_exists: true
          - key: "data_stream/dataset"
            value: "apache"
            add_when: 'contains(/logGroup, "/apache/")'
            overwrite_if_key_exists: true
    # デフォルトのCloudWatchフィールドを削除します。これらはSS4Oフィールドに再マッピングされました:
    - delete_entries:
        with_keys:
          - "logGroup"
          - "logStream"
          - "owner"
    # 非JSONのapacheログを解析するためにGrokパーサーを使用します
    - grok:
        grok_when: '/data_stream/dataset == "apache"'
        match:
          message: ["%{COMMONAPACHELOG_DATATYPED}"]
        target_key: "http"
    # OpenSearchインデックスでのフィールドレベルの検索をサポートするために、ログデータをJSONとして解析しようとします:
    - parse_json:
        # ルートメッセージオブジェクトをaws.cloudtrailに解析して、SS4OログのSS4O標準に一致させます
        source: "message"
        destination: "aws/cloudtrail"
        parse_when: '/data_stream/dataset == "cloudtrail"'
        tags_on_failure: ["json_parse_fail"]
    - parse_json:
        # Lambda関数ログの場合、可能であればルートメッセージオブジェクトをJSONとして解析します - 非JSONのログ関数データを検索可能なフィールドとしてキャプチャするためにGrokサポートを設定することもできます
        source: "message"
        destination: "aws/lambda"
        parse_when: '/data_stream/dataset == "lambda"'
        tags_on_failure: ["json_parse_fail"]
    - parse_json:
        # 一般ログの場合、可能であればルートメッセージオブジェクトをJSONとして解析します
        source: "message"
        destination: "body"
        parse_when: '/data_stream/dataset == "general"'
        tags_on_failure: ["json_parse_fail"]

  sink:
    - opensearch:
        # AWS OpenSearch Serviceドメインエンドポイントを提供します
        hosts: ["https://****************.ap-northeast-1.es.amazonaws.com"]
        # ログデータをログコンテキストに応じて異なるターゲットインデックスにルーティングします:
        index: "ss4o_${data_stream/type}-${data_stream/dataset}-${data_stream/namespace}"
        aws:
          # ドメインへのアクセスを持つRole ARNを提供します。このロールはosis-pipelines.amazonaws.comとの信頼関係を持つ必要があります。
          # このロールは、上記のKinesisで使用したロールと同じでなければなりません。
          sts_role_arn: "arn:aws:iam::<account-id>:role/kds-to-es-integration-pipeline-role"
          # ドメインのリージョンを提供します。
          region: "ap-northeast-1"
          # シンクがAmazon OpenSearch Serverlessコレクションである場合、'serverless'フラグを有効にします
          serverless: false